package com.bfxy.disruptor.common;

import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.consume.SingleConsumer;
import com.bfxy.disruptor.common.consume.ConsumeFactory;
import com.bfxy.disruptor.common.consume.EventConsumer;
import com.bfxy.disruptor.common.consume.IConsumerInfo;
import com.bfxy.disruptor.common.event.*;
import com.bfxy.disruptor.common.exception.ExceptionHandler;
import com.bfxy.disruptor.common.exception.ExceptionHandlerSetting;
import com.bfxy.disruptor.common.exception.TimeoutException;
import com.bfxy.disruptor.common.waitStraty.WaitStrategy;

public class Disruptor<T> {
	private final RingBuffer<T> ringBuffer;// 环状数组就是RingBuffer
	/**消费者事件处理线程池**/
	private final Executor executor;
	/**消费者集合**/
	private final ConsumerRepository<T> consumerRepository = new ConsumerRepository<T>();
	private final AtomicBoolean started = new AtomicBoolean(false);//Disruptor是否启动/**Disruptor是否启动标示，只能启动一次**/
	private ExceptionHandler<? super T> exceptionHandler;/**消费者事件异常处理方法**/

	public Disruptor(final EventFactory<T> eventFactory, final int ringBufferSize, final Executor executor) {
		this(RingBuffer.createMultiProducer(eventFactory, ringBufferSize), executor);
	}

	//构造Disruptor时候，初始化ringBuffer和线程池。
	//构造RingBuffer时候，会选择单生产者还是多生产者的Producer，把Producer和事件工程传给RingBuffer。
	public Disruptor(final EventFactory<T> eventFactory, final int ringBufferSize, final Executor executor,
			final ProducerType producerType, final WaitStrategy waitStrategy) {
		// 这4个参数用于构建RingBuffer
		this(RingBuffer.create(producerType, eventFactory, ringBufferSize, waitStrategy), executor);
	}

	private Disruptor(final RingBuffer<T> ringBuffer, final Executor executor) {
		this.ringBuffer = ringBuffer;// bufferSize=4，entries=68个（32、33、34、35有值是OrderEvent就是消息），indexMask=3，p1-p7=0，sequencer
										// com.bfxy.disruptor.common.SingleProducerSequencer@4439f31e。sequencer里面有bufferSize=4,cursor=Sequence,等待策略
		this.executor = executor;// 等待策略
	}

	//注册封装消费者,每一个消费者最终都会被包装成EventConsumer。
	//创建消费者也是在ringbuuuffer里面。
	public EventHandlerGroup<T> handleEventsWith(final EventHandler<? super T>... handlers) {
		return createEventProcessors(new Sequence[0], handlers);//创建消费者,第一个参数恒为一个空数组[]。
	}

	public EventHandlerGroup<T> handleEventsWith(final ConsumeFactory<T>... eventProcessorFactories) {
		final Sequence[] barrierSequences = new Sequence[0];
		return createEventProcessors(barrierSequences, eventProcessorFactories);
	}

	public EventHandlerGroup<T> handleEventsWith(final EventConsumer... processors) {
		for (final EventConsumer processor : processors) {
			consumerRepository.add(processor);
		}
		return new EventHandlerGroup<T>(this, consumerRepository, Util.getSequencesFor(processors));
	}

	@SuppressWarnings("varargs")
	public EventHandlerGroup<T> handleEventsWithWorkerPool(final WorkHandler<T>... workHandlers) {
		return createWorkerPool(new Sequence[0], workHandlers);
	}

	public void handleExceptionsWith(final ExceptionHandler<? super T> exceptionHandler) {
		this.exceptionHandler = exceptionHandler;
	}

	public ExceptionHandlerSetting<T> handleExceptionsFor(final EventHandler<T> eventHandler) {
		return new ExceptionHandlerSetting<T>(eventHandler, consumerRepository);
	}

	@SuppressWarnings("varargs")
	public EventHandlerGroup<T> after(final EventHandler<T>... handlers) {
		final Sequence[] sequences = new Sequence[handlers.length];
		for (int i = 0, handlersLength = handlers.length; i < handlersLength; i++) {
			sequences[i] = consumerRepository.getSequenceFor(handlers[i]);
		}

		return new EventHandlerGroup<T>(this, consumerRepository, sequences);
	}

	public EventHandlerGroup<T> after(final EventConsumer... processors) {
		for (final EventConsumer processor : processors) {
			consumerRepository.add(processor);
		}

		return new EventHandlerGroup<T>(this, consumerRepository, Util.getSequencesFor(processors));
	}

	//发布事件
	public void publishEvent(final EventTranslator<T> eventTranslator) {
		ringBuffer.publishEvent(eventTranslator);
	}

	public <A> void publishEvent(final EventTranslatorOneArg<T, A> eventTranslator, final A arg) {
		ringBuffer.publishEvent(eventTranslator, arg);
	}

	public <A> void publishEvents(final EventTranslatorOneArg<T, A> eventTranslator, final A[] arg) {
		ringBuffer.publishEvents(eventTranslator, arg);
	}
	
	//start方法，最终会运行到消费者的run方法。
	//Disruptor启动时，会调用每个消费者ConsumerInfo(在消费者集合ConsumerRepository中)的start方法，
	//最终会运行到Consuer的run方法。
	public RingBuffer<T> start() {
		final Sequence[] gatingSequences = consumerRepository.getLastSequenceInChain(true);
		ringBuffer.addGatingSequences(gatingSequences);//保存消费者的进度，到 单或多生产者sequencer

		checkOnlyStartedOnce();
		// 有5个对象就会循环5次。消费run起来。
		for (final IConsumerInfo consumerInfo : consumerRepository) {
			consumerInfo.start(executor);
		}

		return ringBuffer;
	}

	public void halt() {
		for (final IConsumerInfo consumerInfo : consumerRepository) {
			consumerInfo.halt();
		}
	}

	public void shutdown() {
		try {
			shutdown(-1, TimeUnit.MILLISECONDS);
		} catch (final TimeoutException e) {
			exceptionHandler.handleOnShutdownException(e);
		}
	}

	public void shutdown(final long timeout, final TimeUnit timeUnit) throws TimeoutException {
		final long timeOutAt = System.currentTimeMillis() + timeUnit.toMillis(timeout);
		while (hasBacklog()) {
			if (timeout >= 0 && System.currentTimeMillis() > timeOutAt) {
				throw TimeoutException.INSTANCE;
			}
			// Busy spin
		}
		halt();
	}

	public RingBuffer<T> getRingBuffer() {
		return ringBuffer;
	}

	public long getCursor() {//游标是cursor的游标，游标在ringbuffer上面，由Disruptor返回。
		return ringBuffer.getCursor();
	}

	public long getBufferSize() {//buffersize也是由ringbuffer返回。
		return ringBuffer.getBufferSize();
	}

	public T get(final long sequence) {
		return ringBuffer.get(sequence);
	}

	public SequenceBarrier getBarrierFor(final EventHandler<T> handler) {
		return consumerRepository.getBarrierFor(handler);
	}

	private boolean hasBacklog() {
		final long cursor = ringBuffer.getCursor();
		for (final Sequence consumer : consumerRepository.getLastSequenceInChain(false)) {
			if (cursor > consumer.get()) {
				return true;
			}
		}
		return false;
	}
	
	//createEventProcessors是包装消费者的重要函数
	//创建消费者是在ringbuffer里面。所有的消费者公用一个栅栏。
	public EventHandlerGroup<T> createEventProcessors(final Sequence[] barrierSequences,//barrierSequences = new Sequence[0]
			final EventHandler<? super T>[] eventHandlers) {//eventHandlers是消费者的个数, eventHandlers = new Consumer1(),
		checkNotStarted();
		//每个消费者有自己的事件序号Sequence。用来保存每个消费者的消费进度
		final Sequence[] processorSequences = new Sequence[eventHandlers.length];
		
		//生产者没有栅栏，只有消费者才有栅栏，并且栅栏里面也有单或者多生产者sequencer。
		//SequenceBarrier其实就是一个ProcessingSequenceBarrier实例,消费者通过SequenceBarrier等待可消费事件。 创建一个与当前RingBuffer的生产者有关的Barrier
		/*栅栏是ringBuffer创建的，ringBuffer调用ringBuffer里面的sequencer（单或多生产者sequencer）来创建的。
		所以栅栏里面有单或多生产者sequencer（只有一个）以及生产者的位置（在单或多生产者sequencer里面）*/
		final SequenceBarrier barrier = ringBuffer.newBarrier(barrierSequences);

		for (int i = 0, eventHandlersLength = eventHandlers.length; i < eventHandlersLength; i++) {
			final EventHandler<? super T> eventHandler = eventHandlers[i];
			// 每次创建一个SingleConsumer，创建一个就开一个线程，单消费者使用SingleConsumer类，多消费者模式使用WorkProcessor
			// 
			final SingleConsumer<T> batchEventProcessor = new SingleConsumer<T>(ringBuffer, barrier,
					eventHandler);//创建一个消费者，默认是SingleConsumer。每个消费者一个栅栏和一个sequence。

			if (exceptionHandler != null) {
				batchEventProcessor.setExceptionHandler(exceptionHandler);
			}

			consumerRepository.add(batchEventProcessor, eventHandler, barrier);//将EventHandler与BatchEventProcessor关联起来
			processorSequences[i] = batchEventProcessor.getSequence();
		}

		if (processorSequences.length > 0) {//将消费者的sequence添加为RingBuffer的gating sequence
			consumerRepository.unMarkEventProcessorsAsEndOfChain(barrierSequences);
		}
		// 返回一个EventHandlerGroup，这个主要是为了DSL服务的，可以先不关心，可以看到DEMO中我们也没有用到这个返回值
		return new EventHandlerGroup<T>(this, consumerRepository, processorSequences);
	}

	public EventHandlerGroup<T> createEventProcessors(final Sequence[] barrierSequences,
			final ConsumeFactory<T>[] processorFactories) {
		final EventConsumer[] eventProcessors = new EventConsumer[processorFactories.length];
		for (int i = 0; i < processorFactories.length; i++) {
			eventProcessors[i] = processorFactories[i].createEventProcessor(ringBuffer, barrierSequences);
		}
		return handleEventsWith(eventProcessors);
	}

	public EventHandlerGroup<T> createWorkerPool(final Sequence[] barrierSequences,
			final WorkHandler<? super T>[] workHandlers) {
		final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(barrierSequences);
		final WorkerPool<T> workerPool = new WorkerPool<T>(ringBuffer, sequenceBarrier, exceptionHandler, workHandlers);
		consumerRepository.add(workerPool, sequenceBarrier);
		return new EventHandlerGroup<T>(this, consumerRepository, workerPool.getWorkerSequences());
	}

	private void checkNotStarted() {
		if (started.get()) {
			throw new IllegalStateException("All event handlers must be added before calling starts.");
		}
	}

	private void checkOnlyStartedOnce() {
		if (!started.compareAndSet(false, true)) {
			throw new IllegalStateException("Disruptor.start() must only be called once.");
		}
	}
}
